package cn.topcodes.tcsf.admin.service.ops;

import cn.topcode.unicorn.utils.*;
import cn.topcode.unicorn.utils.http.HttpResp;
import cn.topcode.unicorn.utils.http.HttpUtil;
import cn.topcodes.tcsf.admin.domain.entity.SystemProperty;
import cn.topcodes.tcsf.admin.service.SystemPropertyService;
import cn.topcodes.tcsf.admin.service.licence.LicenceException;
import cn.topcodes.tcsf.admin.service.licence.LicenceService;
import cn.topcodes.tcsf.admin.service.ops.proto.*;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.net.*;
import java.nio.charset.Charset;
import java.util.Date;
import java.util.concurrent.*;

/**
 * 连接运维平台
 * @author Unicorn
 * @date 2018/6/22
 */
@Component
public class OpsServiceImpl implements ApplicationListener<ContextRefreshedEvent>, OpsService {

    public static final String URL_SUBMIT_LOG = "http://%s:%d/api/server/submit-server-log";

    private Logger logger = LoggerFactory.getLogger(OpsServiceImpl.class);

    private boolean enable = false;

    private boolean submit = false;

    private String opsServerHost = "ops.topcodes.cn";

    private Integer opsServerUdpPort = 10011;

    private Integer opsServerHttpPort = 80;

    private DatagramSocket socket;

    private StatusReportThread statusReportThread;

    private CmdHandleThread cmdHandleThread;

    @Autowired
    private PropertyConfigurer propertyConfigurer;

    @Autowired
    private SystemPropertyService systemPropertyService;

    @Autowired(required = false)
    private StatusReportCallback statusReportCallback;

    @Autowired
    private LicenceService licenceService;

    private ExecutorService submitThreadPool = null;

    /**
     * 基本状态 - 编号
     */
    private String no;

    private String status = StatusReportRequest.STATUS_STARING;

    @Override
    public void setStatus(String status) {
        this.status = status;
    }

    @PostConstruct
    public void init() {
        if(!isConfigEnable()) {
            return;
        }
        opsServerHost = propertyConfigurer.getProperty("ops.server.host", "ops.topcodes.cn");
        try {
            opsServerUdpPort = Integer.parseInt(propertyConfigurer.getProperty("ops.server.udp.port", "10011"));
        }catch (Exception e) {}
        try {
            opsServerHttpPort = Integer.parseInt(propertyConfigurer.getProperty("ops.server.http.port", "80"));
        }catch (Exception e) {}

        submitThreadPool = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(1),
                new ThreadFactoryBuilder().setNameFormat("OpsService-SubmitThread-%d").build(),
                new ThreadPoolExecutor.DiscardPolicy());

        checkSocket();
    }

    public boolean isConfigEnable() {
        try {
            enable = Boolean.parseBoolean(propertyConfigurer.getProperty("ops.server.enable", "false"));
        }catch (Exception e) {}
        return enable;
    }

    public boolean checkSocket() {
        if(socket == null) {
            try {
                socket = new DatagramSocket();

                statusReportThread = new StatusReportThread();
                statusReportThread.setName("OpsService-StatusReportThread");
                statusReportThread.start();

                cmdHandleThread = new CmdHandleThread();
                cmdHandleThread.setName("OpsService-CmdHandleThread");
                cmdHandleThread.start();
                return true;
            } catch (SocketException e) {
                logger.warn("Ops服务启动失败", e);
                stop();
                return false;
            }
        }
        return true;
    }

    @Override
    public void send(OpsUdpBase udpBase) {
        if(enable && checkSocket()) {
            byte[] data = JsonUtil.toJson(udpBase).getBytes(Charset.forName("UTF-8"));
            try {
                DatagramPacket packet = new DatagramPacket(data, data.length, InetAddress.getByName(opsServerHost), opsServerUdpPort);
                socket.send(packet);
            } catch (UnknownHostException e) {
                logger.warn("无法连接ops服务", e);
            } catch (IOException e) {
                logger.warn("上报状态到ops失败", e);
            }
        }
    }

    public void stop() {
        if(submitThreadPool != null) {
            submitThreadPool.shutdown();
            submitThreadPool = null;
        }
        if(statusReportThread != null) {
            statusReportThread.isRun = false;
            statusReportThread.interrupt();
            statusReportThread = null;
        }
        if(cmdHandleThread != null) {
            cmdHandleThread.isRun = false;
            cmdHandleThread.interrupt();
            cmdHandleThread = null;
        }
        if(socket != null) {
            socket.close();
            socket = null;
        }
    }

    @PreDestroy
    public void destroy() {
        this.status = StatusReportRequest.STATUS_STOPING;
        this.reportStatus();

        stop();
    }

    @Override
    public void reportStatus() {
        StatusReportRequest request = null;
        if(statusReportCallback != null) {
            request = statusReportCallback.getStatus();
        }
        if(request == null) {
            request = new StatusReportRequest();
        }
        request.setType(StatusReportRequest.TYPE_STATUS_REPORT);
        request.setNo(getNo());
        request.setStatus(status);

        send(request);
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if(event.getApplicationContext().getParent() != null) {
            try {
                licenceService.validate();
                this.status = StatusReportRequest.STATUS_RUNNING;
            } catch (LicenceException e) {
                this.status = StatusReportRequest.STATUS_UNAUTHORIZED;
            }
            reportStatus();
        }
    }

    class StatusReportThread extends Thread {

        public boolean isRun = true;

        @Override
        public void run() {
            while(isRun) {

                reportStatus();

                try {
                    Thread.sleep(15 * 1000);
                } catch (InterruptedException e) {}
            }
        }
    }

    class CmdHandleThread extends Thread {

        public boolean isRun = true;

        @Override
        public void run() {
            //  30K缓存区
            byte[] buffer = new byte[1024 * 30];
            while(isRun) {
                DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
                try {
                    socket.receive(packet);
                }catch (Exception e) {}

                try {
                    String json = new String(packet.getData(), 0, packet.getLength(), Charset.forName("UTF-8"));
                    if(StringUtil.isNotBlank(json)) {
                        OpsUdpBase base = JsonUtil.parse(json, OpsUdpBase.class);
                        if(base != null) {
                            if(!submit) {
                                submitThreadPool.submit(new Runnable(){
                                    @Override
                                    public void run() {
                                        OpsConnectedEvent opsConnectedEvent = new OpsConnectedEvent();
                                        opsConnectedEvent.setNo(no);
                                        opsConnectedEvent.setOpsServerHost(opsServerHost);
                                        opsConnectedEvent.setOpsServerHttpPort(opsServerHttpPort);
                                        try {
                                            SpringContextHolder.publishEvent(opsConnectedEvent);
                                            submit = true;
                                        }catch (Exception e) {
                                            submit = false;
                                        }
                                    }
                                });
                            }

                            if(base.isAck()) {
                                OpsUdpResponse response = JsonUtil.parse(json, OpsUdpResponse.class);
                                OpsResponseEvent event = new OpsResponseEvent();
                                event.setOpsUdpResponse(response);
                                event.setData(json);
                                event.setIp(packet.getAddress().getHostAddress());
                                event.setPort(packet.getPort());
                                SpringContextHolder.publishEvent(event);
                            }else {
                                OpsUdpRequest request = JsonUtil.parse(json, OpsUdpRequest.class);
                                OpsRequestEvent event = new OpsRequestEvent();
                                event.setOpsUdpRequest(request);
                                OpsUdpResponse opsUdpResponse = new OpsUdpResponse();
                                opsUdpResponse.setAck(true);
                                event.setData(json);
                                event.setIp(packet.getAddress().getHostAddress());
                                event.setPort(packet.getPort());
                                opsUdpResponse.setType(base.getType());
                                try {
                                    SpringContextHolder.publishEvent(event);
                                    opsUdpResponse.setCode(OpsUdpResponse.CODE_OK);
                                }catch (OpsResponseBody e) {
                                    opsUdpResponse.setCode(e.getCode());
                                    opsUdpResponse.setMessage(e.getMessage());
                                }
                                send(opsUdpResponse);
                            }
                        }
                    }
                } catch (Exception e) {
                    logger.warn("Ops处理发生错误", e);
                }
            }
        }
    }

    @Override
    public String getNo() {
        if(StringUtil.isNotBlank(this.no)) {
            return this.no;
        }
        this.no = systemPropertyService.findSysProperty("APP.NODE.NO", String.class, null);
        if(StringUtil.isBlank(this.no)) {
            SystemProperty prop = new SystemProperty();
            prop.setDescription("应用程序节点标识");
            prop.setName("APP.NODE.NO");
            prop.setType("String");
            prop.setValue(UuidUtil.getUUID());
            systemPropertyService.create(prop);
            this.no = prop.getValue();
        }
        return this.no;
    }

    @Override
    public void clearSubmit() {
        this.submit = false;
    }

    @Override
    public String getOpsServerHost() {
        return opsServerHost;
    }

    @Override
    public Integer getOpsServerHttpPort() {
        return opsServerHttpPort;
    }

    @Override
    public boolean isEnable() {
        return enable;
    }

    @Override
    @Async
    public void submitServerLog(String content) {
        SubmitServerLog submitServerLog = new SubmitServerLog();
        submitServerLog.setContent(content);
        submitServerLog.setTime(System.currentTimeMillis());
        submitServerLog.setNo(this.no);
        String url = String.format(URL_SUBMIT_LOG, this.opsServerHost, this.opsServerHttpPort);
        HttpResp resp = HttpUtil.post(submitServerLog, url);
        if(resp != null && resp.getStatusCode() == 200) {
            SubmitResult result = JsonUtil.parse(resp.getBody(), SubmitResult.class);
            if(result != null && "OK".equals(result.getCode())) {
                return;
            }
        }
        logger.warn("节点日志提交ops失败，日志内容：" + content);
    }
}
